package cn.doitedu.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class _07_Query_CumulateWindow {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // source中如果某个partition空闲时间超过1s，则不参与watermark合并
        tenv.getConfig().set("table.exec.source.idle-timeout","1s");


        // 创建数据源的映射表
        tenv.executeSql(
                        " create table action_log(                              "+
                        "   uid int,                                            "+
                        " 	event_id string,                                    "+
                        " 	properties map<string,string>,                      "+
                        " 	action_time bigint,                                 "+
                        " 	pt as proctime(),                                   "+
                        " 	rt as to_timestamp_ltz(action_time,3),              "+
                        " 	watermark for rt as rt                              "+
                        "  ) with (                                             "+
                        "    'connector' = 'kafka',                             "+
                        "    'topic' = 'action-log',                            "+
                        "    'properties.bootstrap.servers' = 'doitedu:9092',   "+
                        "    'properties.group.id' = 'g003',                    "+
                        "    'scan.startup.mode' = 'latest-offset',             "+
                        "    'value.format' = 'json',                           "+
                        "    'value.fields-include' = 'EXCEPT_KEY'              "+
                        " )                                                     "
        );


        // 统计当天0点到当前累计的pv、uv，每1分钟更新一次
        tenv.executeSql(
                         " SELECT                                                                                "+
                         "   window_start,                                                                       "+
                        "    window_end,                                                                         "+
                        "    count(event_id) filter(where event_id='page_load') as pv,                           "+
                        "    count(distinct uid) filter(where event_id = 'page_load') as uv                      "+
                        " FROM TABLE(                                                                            "+
                        "    CUMULATE(TABLE action_log,DESCRIPTOR(rt),INTERVAL '1' MINUTE, INTERVAL '24' HOUR)   "+
                        " )                                                                                      "+
                        " GROUP BY                                                                               "+
                        "    window_start,                                                                       "+
                        "    window_end                                                                          "
        ).print();




    }

}
